package com.example.producer.controller;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.alibaba.fastjson.JSON;
import com.example.common.model.CommonConstant;
import com.example.common.model.Dog;

@RestController
@RequestMapping("/producer")
public class ProducerController {

    // TOPIC名称
    private static final String                TOPIC            = "transTopic";
    // TAG信息
    private static final String                TAG              = "transTag";
    // 创建一个 ThreadLocal 对象，用于存储用户信息
    private static ThreadLocal<Map<Long, Dog>> threadLocalUsers = ThreadLocal.withInitial(HashMap::new);
    @Autowired
    private RocketMQTemplate                   rocketMQTemplate;

    /**
     * 同步发送
     */
    @RequestMapping("/1")
    public String test1() {

        long start = System.currentTimeMillis();

        for (int i = 0; i < 10; i++) {
            Dog dog = buildDog((long) i, "同步消息");

            String msg = JSON.toJSONString(dog);
            SendResult sendResult = rocketMQTemplate.syncSend(CommonConstant.MQ_TOPIC, msg, 10000);
            System.out.println("同步发送成功, msgId: " + sendResult.getMsgId());

            Map<Long, Dog> dogMap = threadLocalUsers.get();
            Dog dog1 = dogMap.get((long) i);

            System.out.println("id: " + dog1 + ", dog: " + JSON.toJSONString(dog1));

        }

        String threadName = Thread.currentThread().getName();

        ThreadLocal<String> a = new ThreadLocal<String>();
        String s = a.get();

        long end = System.currentTimeMillis();

        long duration = (end - start);

        return "发送完毕，用时: " + duration + " ms, + threadName: " + threadName;
    }

    /**
     * 异步发送
     */
    @RequestMapping("/2")
    public String test2() {

        long start = System.currentTimeMillis();

        for (int i = 0; i < 1000; i++) {
            Dog dog = buildDog((long) i, "异步消息");

            // 将用户信息存储到 ThreadLocal 中
            Map<Long, Dog> userMap = threadLocalUsers.get();
            userMap.put(dog.getId(), dog);

            String msg = JSON.toJSONString(dog);
            rocketMQTemplate.asyncSend(CommonConstant.MQ_TOPIC, msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println("异步发送成功, msgId: " + sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable throwable) {

                }
            });
        }

        long end = System.currentTimeMillis();

        long duration = (end - start);

        return "发送完毕，用时: " + duration + " ms";
    }

    /**
     * 一次性发送
     */
    @RequestMapping("/3")
    public String test3() {

        long start = System.currentTimeMillis();

        for (int i = 0; i < 1000; i++) {
            Dog dog = buildDog(null, "一次性消息");

            String msg = JSON.toJSONString(dog);
            rocketMQTemplate.sendOneWay(CommonConstant.MQ_TOPIC, msg);
        }

        long end = System.currentTimeMillis();

        long duration = (end - start);

        return "发送完毕，用时: " + duration + " ms";
    }

    /**
     * 延迟消息
     */
    @RequestMapping("/4")
    public String test4() {

        long start = System.currentTimeMillis();

        for (int i = 0; i < 5; i++) {
            Dog dog = buildDog((long) i, "延迟消息");

            String msg = JSON.toJSONString(dog);

            Message<?> message = MessageBuilder.withPayload(msg).build();

            rocketMQTemplate.syncSend(CommonConstant.MQ_TOPIC, message, 1000, 3);
            System.out.println("发送延迟消息");
        }

        long end = System.currentTimeMillis();

        long duration = (end - start);

        return "发送完毕，用时: " + duration + " ms";
    }

    /**
     * 顺序消息
     */
    @RequestMapping("/5")
    public String test5() {

        long start = System.currentTimeMillis();

        // hashKey用来计算决定消息发送到哪个消息队列 一般是订单ID，产品ID等

        rocketMQTemplate.syncSendOrderly(CommonConstant.MQ_TOPIC, "98456233,创建", "98456233");
        rocketMQTemplate.syncSendOrderly(CommonConstant.MQ_TOPIC, "98456233,支付", "98456233");

        rocketMQTemplate.syncSendOrderly(CommonConstant.MQ_TOPIC, "98456231,创建", "98456231");
        rocketMQTemplate.syncSendOrderly(CommonConstant.MQ_TOPIC, "98456231,支付", "98456231");

        rocketMQTemplate.syncSendOrderly(CommonConstant.MQ_TOPIC, "98456234,创建", "98456234");

        rocketMQTemplate.syncSendOrderly(CommonConstant.MQ_TOPIC, "98456232,创建", "98456232");
        rocketMQTemplate.syncSendOrderly(CommonConstant.MQ_TOPIC, "98456232,支付", "98456232");
        rocketMQTemplate.syncSendOrderly(CommonConstant.MQ_TOPIC, "98456232,完成", "98456232");

        rocketMQTemplate.syncSendOrderly(CommonConstant.MQ_TOPIC, "98456233,完成", "98456233");

        rocketMQTemplate.syncSendOrderly(CommonConstant.MQ_TOPIC, "98456234,支付", "98456234");
        rocketMQTemplate.syncSendOrderly(CommonConstant.MQ_TOPIC, "98456234,完成", "98456234");

        rocketMQTemplate.syncSendOrderly(CommonConstant.MQ_TOPIC, "98456231,完成", "98456231");

        long end = System.currentTimeMillis();

        long duration = (end - start);

        return "发送完毕，用时: " + duration + " ms";
    }

    @RequestMapping("/6")
    public String test6() {

        // 生成事务ID
        String transactionId = UUID.randomUUID().toString().replace("-", "");
        //        log.info("【发送半消息】transactionId={}", transactionId);
        String transKeys = "transKey";

        String msg = "事务消息测试, " + LocalDateTime.now();

        // 发送事务消息
        TransactionSendResult sendResult = rocketMQTemplate.sendMessageInTransaction(CommonConstant.MQ_TX_GROUP, TOPIC, MessageBuilder.withPayload(msg) //
            .setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId) //
            .setHeader(RocketMQHeaders.KEYS, transKeys) // 相比于使用"KEYS"，使用封装常量更不易出错
            .build(), //
            msg);
        System.out.println(sendResult);
        System.out.println(sendResult.getLocalTransactionState());
        //        System.out.println("【发送半消息】sendResult={}" + msg);

        return "事务消息发送完毕";
    }

    private Dog buildDog(Long id, String name) {

        long currMill = LocalDateTime.now().toInstant(ZoneOffset.ofHours(8)).toEpochMilli();

        Dog dog = new Dog() //
            .setId(currMill) //
            .setName(name) //
            .setSize(BigDecimal.ONE) //
        //            .setCreateTime(LocalDateTime.now()) //
        ;
        if (id != null) {
            dog.setId(id);
        }

        System.out.print("dog: " + JSON.toJSONString(dog) + " ");

        return dog;
    }

}
